package com.lhl.kafka.kafkaconsumer;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @author LHL
 * @date 2019/4/10
 */
@Component
public class KafkaConsumer1 {
    public KafkaConsumer1() {
    }

    private static final Logger log = LoggerFactory.getLogger(KafkaConsumer1.class);

//    @KafkaListener(topics = "demo1",groupId = "group1")
//    public void consumer(ConsumerRecord<String, String> record){
//        log.info("======>topic:{},value:{},partition:{},thread:{}",record.topic(),record.value(),record.partition(),Thread.currentThread().getName());
//    }
//
//    @KafkaListener(topics = "demo1",groupId = "group1")
//    public void consumer1(ConsumerRecord<String, String> record) {
//
//        log.info("------->topic:{},value:{},partition:{},thread:{}", record.topic(), record.value(),record.partition(),Thread.currentThread().getName());
//    }


//    @KafkaListener(topics = "demo1",groupId = "group1",containerFactory = "kafkaListenerContainerFactory")
//    public void consumerBatch(List<ConsumerRecord<String, String>> records) {
//        log.info("===============start batch receive================");
//        records.forEach(record->{
//            log.info("!!!!!!!!!!!!!!>topic:{},valur:{},offset:{}", record.topic(), record.value(), record.offset());
//        });
//    }
    @KafkaListener(topics = "demo1",groupId = "group1",containerFactory = "kafkaListenerContainerFactory")
    public void consumerBatch(List<ConsumerRecord<String, String>> records, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
        log.info("===============start batch receive================");
        records.forEach(record->{

            if (record.value().equals("3333")){
                log.info(",,,,,,,,,,,>topic:{},valur:{},offset:{},partition:{}", record.topic(), record.value(), record.offset(),record.partition());
                return;
//                throw new RuntimeException();
            }
//            acknowledgment.acknowledge();
            log.info("!!!!!!!!!!!!!!>topic:{},valur:{},offset:{},partition:{}", record.topic(), record.value(), record.offset(),record.partition());
        });
    }

    }

